package com.at.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/27 8:05 PM
 */
object Spark01_SparkSql_Basic {

  def testDataFrame(spark: SparkSession) = {
    import spark.implicits._
    val df = spark.read.json("datas/user.json")
    //    df.show()
    // DataFrame => SQL
    df.createOrReplaceTempView("user")
    //    spark.sql("select * from user").show()
    // DataFrame => DSL
    // 在使用DataFrame时，如果涉及到转换操作，需要进入转换规则

    df.select("age", "name").show()
    df.select($"age" + 1).show()
    df.select('age + 1).show()
  }

  def testDataSet(spark: SparkSession) = {
    import spark.implicits._
    // DataFrame其实是特定泛型的DataSet
    val seq = Seq(1, 2, 3, 4)
    val ds = seq.toDS()
    ds.show()
  }

  def convert(spark: SparkSession) = {
    import spark.implicits._
    val rdd = spark.sparkContext.makeRDD(List((1, "zs", 30), (2, "ls", 340)))

    // RDD <=> DataFrame
    val df = rdd.toDF("id", "name", "age")
    val rowRDD = df.rdd

    // DataFrame <=> DataSet
    val ds = df.as[User]
    val df1 = ds.toDF()

    // RDD <=> DataSet
    val ds1 = rdd.map {
      case (id, name, age) => {
        User(id, name, age)
      }
    }.toDS()
    val userRDD = ds1.rdd
  }

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("operator")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    // TODO DataFrame
    //    testDataFrame(spark)

    // TODO DataSet
    //    testDataSet(spark)

    convert(spark)

    spark.close();
  }

  case class User(id: Int, name: String, age: Int)
}
